package com.serotonin.messaging;

import java.io.IOException;

import com.serotonin.io.StreamUtils;
import com.serotonin.modbus4j.ip.xa.XaMessageResponse;
import com.serotonin.util.queue.ByteQueue;

public class MessageControl implements DataConsumer {
	private static int DEFAULT_RETRIES = 2;
	private static int DEFAULT_TIMEOUT = 60000;
	public boolean DEBUG;
	private Transport transport;
	private MessageParser messageParser;
	private RequestHandler requestHandler;
	private MessagingExceptionHandler exceptionHandler;
	private int retries;
	private int timeout;
	private final WaitingRoom waitingRoom;
	private final ByteQueue dataBuffer;

	public MessageControl() {
		this.DEBUG = false;

		this.exceptionHandler = new DefaultMessagingExceptionHandler();
		this.retries = DEFAULT_RETRIES;
		this.timeout = DEFAULT_TIMEOUT;

		this.waitingRoom = new WaitingRoom();
		this.dataBuffer = new ByteQueue();
	}

	public void start(Transport transport, MessageParser messageParser,
			RequestHandler handler) throws IOException {
		this.transport = transport;
		this.messageParser = messageParser;
		this.requestHandler = handler;
		transport.setConsumer(this);
	}

	public void close() {
		this.transport.removeConsumer();
	}

	public void setExceptionHandler(MessagingExceptionHandler exceptionHandler) {
		if (exceptionHandler == null)
			this.exceptionHandler = new DefaultMessagingExceptionHandler();
		else
			this.exceptionHandler = exceptionHandler;
	}

	public int getRetries() {
		return this.retries;
	}

	public void setRetries(int retries) {
		this.retries = retries;
	}

	public int getTimeout() {
		return this.timeout;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	public IncomingResponseMessage send(OutgoingRequestMessage request)
			throws IOException {
		return send(request, this.timeout, this.retries);
	}

	public IncomingResponseMessage send(OutgoingRequestMessage request,
			int timeout, int retries) throws IOException {
		byte[] data = request.getMessageData();
		if (this.DEBUG)
			System.out.println("MessagingControl.send: "
					+ StreamUtils.dumpMessage(data));

		IncomingResponseMessage response = null;
		if (request.expectsResponse()) {
			WaitingRoomKey key = request.getWaitingRoomKey();
			try {
				this.waitingRoom.enter(key);
				do {
					write(data);

					response = this.waitingRoom.getResponse(key, timeout);

					if ((this.DEBUG) && (response == null))
						System.out.println("Timeout waiting for response");

					if (response != null)
						break;
				} while (retries-- > 0);
			} finally {
				this.waitingRoom.leave(key);
			}

			if (response == null)
				System.out.println();
			 //throw new TimeoutException("request=" + request);
		} else {
			//System.out.println("不需要回复");
			write(data);
		}
		return response;
	}

	public void send(OutgoingResponseMessage response) throws IOException {
		write(response.getMessageData());
	}

	public void data(byte[] b, int len) {
		if (this.DEBUG)
			System.out.println("MessagingConnection.read: "
					+ StreamUtils.dumpMessage(b, 0, len));

		this.dataBuffer.push(b, 0, len);
		try {
			this.dataBuffer.mark();

			IncomingMessage message = this.messageParser
					.parseMessage(this.dataBuffer);

			if (message == null) {
				this.dataBuffer.reset();
				return;
			}

			if (message instanceof IncomingRequestMessage) {
				if (this.requestHandler != null) {
					OutgoingResponseMessage response = this.requestHandler
							.handleRequest((IncomingRequestMessage) message);

					if (response != null)
						send(response);
				}
			} else {
				try {
					this.waitingRoom
							.response((IncomingResponseMessage) message);
				} catch (Exception e) {
					// e.printStackTrace();
				}
			}
		} catch (Exception e) {
			// this.exceptionHandler.receivedException(e);
		}
	}

	private void write(byte[] data) throws IOException {
		synchronized (this.transport) {
			this.transport.write(data);
		}
	}

	public void handleIOException(IOException e) {
		// this.exceptionHandler.receivedException(e);
	}

}
